package com.gugee.ins.data.blogger.es.listener;

import com.gugee.ins.data.blogger.es.config.BloggerToElasticsearchAppConfig;
import com.gugee.ins.data.blogger.es.core.BloggerActiveFollowerDataToEsService;
import com.gugee.ins.data.blogger.es.core.BloggerMainUpdateEsService;
import com.gugee.ins.data.blogger.es.core.BloggerMediaAdToEsService;
import com.gugee.ins.data.blogger.es.core.BloggerRealFollowerToEsService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class BloggerToEsListener {

    @Autowired
    BloggerToElasticsearchAppConfig bloggerToElasticsearchAppConfig;

    @Autowired
    BloggerMainUpdateEsService bloggerMainUpdateEsService;

    @Autowired
    BloggerRealFollowerToEsService bloggerRealFollowerToEsService;

    @Autowired
    BloggerActiveFollowerDataToEsService bloggerActiveFollowerDataToEsService;

    @Autowired
    BloggerMediaAdToEsService bloggerMediaAdToEsService;

    private BloggerMainUpdateEs bloggerMainUpdateEs = new BloggerMainUpdateEs();

    private BloggerMediaAdToEs bloggerMediaAdToEs = new BloggerMediaAdToEs();

    private BloggerActiveFollowerDataToEs bloggerActiveFollowerDataToEs = new BloggerActiveFollowerDataToEs();

    private BloggerRealFollowerToEs bloggerRealFollowerToEs = new BloggerRealFollowerToEs();


    /**
     * 大播基础数据更新到ES
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerUpdateEs.id}",topics = {"${kafka.topics.insMBloggerUpdateEs.online}"},containerFactory = "factoryInsBloggerEsAck")
    public void bloggerMainUpdateEsListen(List<InsBloggerMessage> messages, Acknowledgment ack) {
        bloggerMainUpdateEs.handle("BloggerMainUpdateEs", messages,bloggerToElasticsearchAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 大播主广告信息更新到ES
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBMediaAdEs.id}",topics = {"${kafka.topics.insMBMediaAdEs.online}"},containerFactory = "factoryInsBloggerEsAck")
    public void bloggerMediaAdToEsListen(List<InsBloggerMessage> messages, Acknowledgment ack) {
        bloggerMediaAdToEs.handle("bloggerMediaAdToEs", messages,bloggerToElasticsearchAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 大播真实粉丝更新到ES
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerRealFollowerEs.id}",topics = {"${kafka.topics.insMBloggerRealFollowerEs.online}"},containerFactory = "factoryInsBloggerEsAck")
    public void bloggerRealFollowerToEsListen(List<InsBloggerMessage> messages, Acknowledgment ack) {
        bloggerRealFollowerToEs.handle("BloggerRealFollowerToEs", messages,bloggerToElasticsearchAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    /**
     * 大播基础活跃粉丝数据更新到ES
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMBloggerActiveFollowerDataToEs.id}",topics = {"${kafka.topics.insMBloggerActiveFollowerDataToEs.online}"},containerFactory = "factoryInsBloggerEsAck")
    public void bloggerActiveFollowerDataToEsListen(List<InsBloggerMessage> messages, Acknowledgment ack) {
        bloggerActiveFollowerDataToEs.handle("BloggerActiveFollowerDataToEs", messages,bloggerToElasticsearchAppConfig.getAcceptMsgClMaxLimitTime(),ack);
    }

    class BloggerMainUpdateEs extends AbstractGugeeConsumerHandler<InsBloggerMessage> {

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            bloggerMainUpdateEsService.handler(messages);
        }
    }

    class BloggerMediaAdToEs extends AbstractGugeeConsumerHandler<InsBloggerMessage> {

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            bloggerMediaAdToEsService.handler(messages);
        }
    }

    class BloggerActiveFollowerDataToEs extends AbstractGugeeConsumerHandler<InsBloggerMessage> {

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            bloggerActiveFollowerDataToEsService.handler(messages);
        }
    }

    class BloggerRealFollowerToEs extends AbstractGugeeConsumerHandler<InsBloggerMessage> {

        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            bloggerRealFollowerToEsService.handler(messages);
        }
    }

}
